use std::mem;
use std::{cell::Cell, ffi::CString, marker::PhantomData};

use super::enums::AtomicMode;
use super::message::MessageHeader;
use super::share_memory::OsIpcSharedMemory;

use super::ipc_bridge::{self, OsIpcChannel};
use super::enums::WinIpcError;
use super::message::OutOfBandMessage;
use super::utils::{self, CURRENT_PROCESS_HANDLE, CURRENT_PROCESS_ID};
use super::win_handle::WinHandle;
use crate::win32_trace;

use uuid::Uuid;
use windows::core::Error as WinError;
use windows::Win32::Storage::FileSystem;
use windows::Win32::System::Threading::PROCESS_DUP_HANDLE;



#[derive(Debug)]
#[cfg_attr(feature = "windows-shared-memory-equality", derive(PartialEq))]
pub struct OsIpcSender {
    handle: WinHandle,
    // Make sure this is `!Sync`, to match `mpsc::Sender`; and to discourage sharing references.
    //
    // (Rather, senders should just be cloned, as they are shared internally anyway --
    // another layer of sharing only adds unnecessary overhead...)
    nosync_marker: PhantomData<Cell<()>>,
}

impl Clone for OsIpcSender {
    fn clone(&self) -> OsIpcSender {
        OsIpcSender::from_handle(super::utils::dup_handle(&self.handle).unwrap())
    }
}

impl OsIpcSender {
    pub fn connect(name: String) -> Result<OsIpcSender, WinError> {
        let pipe_name = super::utils::make_pipe_name(&Uuid::parse_str(&name).unwrap());
        OsIpcSender::connect_named(&pipe_name)
    }

    pub fn get_max_fragment_size() -> usize {
        super::utils::MAX_FRAGMENT_SIZE
    }

    pub(super) fn from_handle(handle: WinHandle) -> OsIpcSender {
        OsIpcSender {
            handle: handle,
            nosync_marker: PhantomData,
        }
    }

    /// Connect to a pipe server.
    pub(super) fn connect_named(pipe_name: &CString) -> Result<OsIpcSender, WinError> {
        unsafe {
            let handle = FileSystem::CreateFileA(
                windows::core::PCSTR::from_raw(pipe_name.as_ptr() as *const u8),
                FileSystem::FILE_GENERIC_WRITE.0,
                FileSystem::FILE_SHARE_MODE(0),
                None, // lpSecurityAttributes
                FileSystem::OPEN_EXISTING,
                FileSystem::FILE_ATTRIBUTE_NORMAL,
                None,
            )?;

            win32_trace!("[c {:?}] connect_to_server success", handle);

            Ok(OsIpcSender::from_handle(WinHandle::new(handle)))
        }
    }

    fn get_pipe_server_process_id(&self) -> Result<u32, WinError> {
        unsafe {
            let mut server_pid = 0;
            if windows::Win32::System::Pipes::GetNamedPipeServerProcessId(self.handle.as_raw(), &mut server_pid) == false {
                return Err(WinError::from_win32());
            }
            Ok(server_pid)
        }
    }

    fn get_pipe_server_process_handle_and_pid(&self) -> Result<(WinHandle, u32), WinError> {
        unsafe {
            let server_pid = self.get_pipe_server_process_id()?;
            if server_pid == *CURRENT_PROCESS_ID {
                return Ok((WinHandle::new(CURRENT_PROCESS_HANDLE.as_raw()), server_pid));
            }

            let raw_handle = windows::Win32::System::Threading::OpenProcess(PROCESS_DUP_HANDLE, false, server_pid)?;

            Ok((WinHandle::new(raw_handle), server_pid))
        }
    }

    fn needs_fragmentation(data_len: usize, oob: &OutOfBandMessage) -> bool {
        let oob_size = if oob.needs_to_be_sent() {
            bincode::serialized_size(oob).unwrap()
        } else {
            0
        };

        // make sure we don't have too much oob data to begin with
        assert!(
            (oob_size as usize) <= (utils::PIPE_BUFFER_SIZE - mem::size_of::<MessageHeader>()),
            "too much oob data"
        );

        let bytes_left_for_data =
            (utils::PIPE_BUFFER_SIZE - mem::size_of::<MessageHeader>()) - (oob_size as usize);
        data_len >= bytes_left_for_data
    }

    /// An internal-use-only send method that sends just raw data, with no header.
    fn send_raw(&self, data: &[u8]) -> Result<(), WinError> {
        win32_trace!(
            "[c {:?}] writing {} bytes raw to (pid {}->{})",
            self.handle.as_raw(),
            data.len(),
            *CURRENT_PROCESS_ID,
            self.get_pipe_server_process_id()?
        );

        // Write doesn't need to be atomic,
        // since the pipe is exclusive for this message,
        // so we don't have to fear intermixing with parts of other messages.
        utils::write_buf(&self.handle, data, AtomicMode::Nonatomic)
    }

    pub fn send(
        &self,
        data: &[u8],
        ports: Vec<OsIpcChannel>,
        shared_memory_regions: Vec<OsIpcSharedMemory>,
    ) -> Result<(), WinIpcError> {
        // We limit the max size we can send here; we can fix this
        // just by upping the header to be 2x u64 if we really want
        // to.
        assert!(data.len() <= u32::max_value() as usize);

        let (server_h, server_pid) = if !shared_memory_regions.is_empty() || !ports.is_empty() {
            self.get_pipe_server_process_handle_and_pid()?
        } else {
            (WinHandle::invalid(), 0)
        };

        let mut oob = OutOfBandMessage::new(server_pid);

        for ref shmem in shared_memory_regions {
            // shmem.handle, shmem.length
            let mut remote_handle = utils::dup_handle_to_process(&shmem.handle, &server_h)?;
            oob.shmem_handles
                .push((remote_handle.take_raw().0, shmem.length as u64));
        }

        for port in ports {
            match port {
                OsIpcChannel::Sender(s) => {
                    let mut raw_remote_handle = utils::move_handle_to_process(s.handle, &server_h)?;
                    oob.channel_handles.push(raw_remote_handle.take_raw().0);
                },
                OsIpcChannel::Receiver(r) => {
                    if r.prepare_for_transfer()? == false {
                        panic!("Sending receiver with outstanding partial read buffer, noooooo!  What should even happen?");
                    }

                    let handle = r.reader.into_inner().handle.take();
                    let mut raw_remote_handle = utils::move_handle_to_process(handle, &server_h)?;
                    oob.channel_handles.push(raw_remote_handle.take_raw().0);
                },
            }
        }

        // Do we need to fragment?
        let big_data_sender: Option<OsIpcSender> =
            if OsIpcSender::needs_fragmentation(data.len(), &oob) {
                // We need to create a channel for the big data
                let (sender, receiver) = ipc_bridge::channel()?;

                let (server_h, server_pid) = if server_h.is_valid() {
                    (server_h, server_pid)
                } else {
                    self.get_pipe_server_process_handle_and_pid()?
                };

                // Put the receiver in the OOB data
                let handle = receiver.reader.into_inner().handle.take();
                let mut raw_receiver_handle = utils::move_handle_to_process(handle, &server_h)?;
                oob.big_data_receiver_handle =
                    Some((raw_receiver_handle.take_raw().0, data.len() as u64));
                oob.target_process_id = server_pid;

                Some(sender)
            } else {
                None
            };

        // If we need to send OOB data, serialize it
        let mut oob_data: Vec<u8> = vec![];
        if oob.needs_to_be_sent() {
            oob_data = bincode::serialize(&oob).unwrap();
        }

        let in_band_data_len = if big_data_sender.is_none() {
            data.len()
        } else {
            0
        };
        let header = MessageHeader {
            data_len: in_band_data_len as u32,
            oob_len: oob_data.len() as u32,
        };
        let full_in_band_len = header.total_message_bytes_needed();
        assert!(full_in_band_len <= utils::PIPE_BUFFER_SIZE);
        let mut full_message = Vec::<u8>::with_capacity(full_in_band_len);

        {
            let header_bytes = unsafe {
                std::slice::from_raw_parts(&header as *const _ as *const u8, mem::size_of_val(&header))
            };
            full_message.extend_from_slice(header_bytes);
        }

        if big_data_sender.is_none() {
            full_message.extend_from_slice(&*data);
            full_message.extend_from_slice(&*oob_data);
            assert!(full_message.len() == full_in_band_len);

            // Write needs to be atomic, since otherwise concurrent sending
            // could result in parts of different messages getting intermixed,
            // and the receiver would not be able to extract the individual messages.
            utils::write_buf(&self.handle, &*full_message, AtomicMode::Atomic)?;
        } else {
            full_message.extend_from_slice(&*oob_data);
            assert!(full_message.len() == full_in_band_len);

            utils::write_buf(&self.handle, &*full_message, AtomicMode::Atomic)?;
            big_data_sender.unwrap().send_raw(data)?;
        }

        Ok(())
    }
}